/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.dolphinscheduler.dao.repository.impl;

import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.OrderType;
import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.Bill;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.ProcessInstanceMap;
import org.apache.dolphinscheduler.dao.entity.ProductOrder;
import org.apache.dolphinscheduler.dao.entity.ProductOrderItem;
import org.apache.dolphinscheduler.dao.mapper.BillMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.ProductOrderItemMapper;
import org.apache.dolphinscheduler.dao.mapper.ProductOrderMapper;
import org.apache.dolphinscheduler.dao.repository.BaseDao;
import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.plugin.task.api.model.DateInterval;

import org.apache.commons.collections.CollectionUtils;

import java.math.BigDecimal;
import java.util.Date;
import java.util.List;

import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;

import com.google.common.collect.Lists;

@Slf4j
@Repository
public class ProcessInstanceDaoImpl extends BaseDao<ProcessInstance, ProcessInstanceMapper>
        implements
            ProcessInstanceDao {

    @Autowired
    private ProcessInstanceMapMapper processInstanceMapMapper;

    @Autowired
    private ProductOrderMapper productOrderMapper;

    @Autowired
    private ProductOrderItemMapper productOrderItemMapper;

    @Autowired
    private BillMapper billMapper;

    public ProcessInstanceDaoImpl(@NonNull ProcessInstanceMapper processInstanceMapper) {
        super(processInstanceMapper);
    }

    @Override
    public void upsertProcessInstance(@NonNull ProcessInstance processInstance) {
        if (processInstance.getId() != null) {
            updateById(processInstance);
            updateProductOrder(processInstance);
        } else {
            insert(processInstance);
            saveProductOrder(processInstance);
        }
    }

    /**
     * find last scheduler process instance in the date interval
     *
     * @param definitionCode definitionCode
     * @param dateInterval   dateInterval
     * @return process instance
     */
    @Override
    public ProcessInstance queryLastSchedulerProcessInterval(Long definitionCode, DateInterval dateInterval,
                                                             int testFlag) {
        return mybatisMapper.queryLastSchedulerProcess(definitionCode,
                dateInterval.getStartTime(),
                dateInterval.getEndTime(),
                testFlag);
    }

    /**
     * find last manual process instance interval
     *
     * @param definitionCode process definition code
     * @param dateInterval   dateInterval
     * @return process instance
     */
    @Override
    public ProcessInstance queryLastManualProcessInterval(Long definitionCode, DateInterval dateInterval,
                                                          int testFlag) {
        return mybatisMapper.queryLastManualProcess(definitionCode,
                dateInterval.getStartTime(),
                dateInterval.getEndTime(),
                testFlag);
    }

    /**
     * query first schedule process instance
     *
     * @param definitionCode definitionCode
     * @return process instance
     */
    @Override
    public ProcessInstance queryFirstScheduleProcessInstance(Long definitionCode) {
        return mybatisMapper.queryFirstScheduleProcessInstance(definitionCode);
    }

    /**
     * query first manual process instance
     *
     * @param definitionCode definitionCode
     * @return process instance
     */
    @Override
    public ProcessInstance queryFirstStartProcessInstance(Long definitionCode) {
        return mybatisMapper.queryFirstStartProcessInstance(definitionCode);
    }

    @Override
    public ProcessInstance querySubProcessInstanceByParentId(Integer processInstanceId, Integer taskInstanceId) {
        ProcessInstance processInstance = null;
        ProcessInstanceMap processInstanceMap =
                processInstanceMapMapper.queryByParentId(processInstanceId, taskInstanceId);
        if (processInstanceMap == null || processInstanceMap.getProcessInstanceId() == 0) {
            return processInstance;
        }
        processInstance = queryById(processInstanceMap.getProcessInstanceId());
        return processInstance;
    }

    private void saveProductOrder(ProcessInstance processInstance) {
         log.info("============= create product order start =============");
         log.info(JSONUtils.toJsonString(processInstance));
        if (processInstance.getId() == 0) {
            log.error("create product order error, id is null");
            return;
        }
        int runTimes = processInstance.getRunTimes();
        Date now = DateUtils.getCurrentDate();
        if (runTimes > 1) {
            // 重新启动如果上次的订单状态为执行中，修改为作废
            ProductOrder productOrder =
                    productOrderMapper.selectProductOrderByProcessInstanceId(processInstance.getId(), runTimes - 1);
            if (productOrder != null
                    && productOrder.getTaskStatus().intValue() == WorkflowExecutionStatus.RUNNING_EXECUTION.getCode()) {
                BigDecimal zero = new BigDecimal(0);
                productOrder.setAmount(zero);
                productOrder.setRealAmount(zero);
                productOrder.setDiscountAmount(zero);
                productOrder.setTaskStatus(2);
                productOrder.setPayStatus(2); // 支付状态 0=待支付,1=已支付,2=作废
                productOrder.setPayTime(now);
                productOrder.setUserId(processInstance.getExecutorId());
                productOrder.setEndTime(now);
                productOrderMapper.updateState(productOrder);
            }
        }

        if ((processInstance.getCommandType().equals(CommandType.RECOVER_SUSPENDED_PROCESS)
                || processInstance.getCommandType().equals(CommandType.RECOVER_WAITING_THREAD))
                && runTimes > 1) {
            runTimes--;
            String[] historyCmds = processInstance.getHistoryCmd().split(",");
            if (historyCmds != null && historyCmds.length > 1) {
                String lastCmd = historyCmds[historyCmds.length - 2];
                if (lastCmd.equals("STOP")) {
                    runTimes++;
                }
            }
        }
        ProductOrder productOrder =
                productOrderMapper.selectProductOrderByProcessInstanceId(processInstance.getId(), runTimes);
        if (productOrder != null && productOrder.getId() != null) {
            log.info("productOrder is not null, to update ");
            updateProductOrder(processInstance);
            return;
        }

        // 创建订单
        productOrder = new ProductOrder();
        productOrder.setNo(processInstance.getId() + "-" + System.currentTimeMillis());
        productOrder.setOrderType(OrderType.TASK.getCode()); // 订单类型 0=任务订单,1=模型订单
        productOrder.setProcessInstanceId(processInstance.getId());
        productOrder.setProcessInstanceName(processInstance.getName());
        productOrder.setUserId(processInstance.getExecutorId());
        // productOrder.setPayModel(0); //付费模式 0=后付费
//        productOrder.setTaskStatus(-1); //运行状态 0=运行中,1=运行完成,2=运行失败
         productOrder.setPayStatus(-1); //支付状态 0=待支付,1=已支付,2=作废
        productOrder.setStartTime(processInstance.getStartTime());
        productOrder.setRunTimes(processInstance.getRunTimes());
        productOrder.setUpdateTime(now);
        productOrder.setCreateTime(now);
        productOrderMapper.insert(productOrder);
        // log.info("============= create product order success =============");
    }

    public void updateProductOrder(ProcessInstance processInstance) {
        log.info("============= update product order start =============");
        log.info(JSONUtils.toJsonString(processInstance));
        if (processInstance.getId() == 0) {
            log.error("update product order error, id is null");
            return;
        }

        if(processInstance.getState().equals(WorkflowExecutionStatus.DELAY_EXECUTION)){
            log.info("processInstance state is delay execution");
            return;
        }

        if (processInstance.getState().equals(WorkflowExecutionStatus.READY_PAUSE)) {
            log.info("processInstance state is ready pause");
            return;
        }
        if (processInstance.getState().equals(WorkflowExecutionStatus.READY_STOP)) {
            log.info("processInstance state is ready stop");
            return;
        }
        int runTimes = processInstance.getRunTimes();
        if ((processInstance.getCommandType().equals(CommandType.RECOVER_SUSPENDED_PROCESS)
                || processInstance.getCommandType().equals(CommandType.RECOVER_WAITING_THREAD))
                && runTimes > 1) {
            runTimes--;
            String[] historyCmds = processInstance.getHistoryCmd().split(",");
            if (historyCmds != null && historyCmds.length > 1) {
                String lastCmd = historyCmds[historyCmds.length - 2];
                if (lastCmd.equals("STOP")) {
                    runTimes++;
                }
            }
        }
        ProductOrder productOrder =
                productOrderMapper.selectProductOrderByProcessInstanceId(processInstance.getId(), runTimes);
        if (productOrder == null || productOrder.getId() == null) {
            // 重新执行需要新生成订单
            log.info("productOrder is null, to save");
            saveProductOrder(processInstance);
            return;
        }
        if (!productOrder.getTaskStatus().equals(0)) {
            log.info("productOrder is successed or failured");
            return;
        }
        // 修改订单状态
        productOrder.setProcessInstanceId(processInstance.getId());
        if (processInstance.getStartTime() != null) {
            productOrder.setStartTime(processInstance.getStartTime());
        }
        Date now = new Date();
        BigDecimal zero = new BigDecimal(0);
        productOrder.setAmount(zero);
        productOrder.setRealAmount(zero);
        productOrder.setDiscountAmount(zero);
        //收支明细修改为支付后生成
//        List<Bill> bills = Lists.newArrayList();
        if (processInstance.getState().equals(WorkflowExecutionStatus.SUCCESS)) {
            // || processInstance.getState().equals(WorkflowExecutionStatus.FORCED_SUCCESS)) {
            ProductOrderItem item = productOrderItemMapper.queryTotalAmountByProductOrderId(productOrder.getId());
            if (item != null && item.getAmount() != null) {
                productOrder.setAmount(item.getAmount());
            }
            if (item != null && item.getRealAmount() != null) {
                productOrder.setRealAmount(item.getRealAmount());
            }
            if (item != null && item.getDiscountAmount() != null) {
                productOrder.setDiscountAmount(item.getDiscountAmount());
            }
            productOrder.setTaskStatus(1);
            productOrder.setPayStatus(0); // 支付状态 0=待支付,1=已支付,2=作废
//            productOrder.setPayTime(now);
//            if (productOrder.getRealAmount().intValue() > 0) {
//                bills.addAll(buildBillList(productOrder, processInstance));
//            }
        } else if (processInstance.getState().equals(WorkflowExecutionStatus.FAILURE)
                || processInstance.getState().equals(WorkflowExecutionStatus.PAUSE)
                || processInstance.getState().equals(WorkflowExecutionStatus.STOP)) {
            // || processInstance.getState().equals(WorkflowExecutionStatus.KILL)) {
            productOrder.setTaskStatus(2);
            productOrder.setPayStatus(2); // 支付状态 0=待支付,1=已支付,2=作废
            productOrder.setPayTime(now);
        }
        productOrder.setUserId(processInstance.getExecutorId());
        productOrder.setStartTime(processInstance.getStartTime());
        productOrder.setEndTime(processInstance.getEndTime());
        productOrderMapper.updateState(productOrder);
        // 保存收支明细
//        if (CollectionUtils.isNotEmpty(bills)) {
//            log.info("insert " + JSONUtils.toJsonString(bills) + "bill list");
//            for (Bill bill : bills) {
//                billMapper.insert(bill);
//            }
//        }
         log.info("============= update product order success =============");
    }

    /**
     * 构建收支明细*
     * @param productOrder
     * @param processInstance
     */
    private List<Bill> buildBillList(ProductOrder productOrder, ProcessInstance processInstance) {
        List<Bill> bills = Lists.newArrayList();
        String billNo = "" + System.currentTimeMillis() + (int) (Math.random() * 100);
        Bill billIncome = new Bill();
        billIncome.setUserId(1);
        billIncome.setBillType(1);
        billIncome.setTradeType(0);
        billIncome.setBillNumber(productOrder.getRealAmount());
        billIncome.setOrderNo(productOrder.getNo());
        billIncome.setBillNo(billNo);

        Bill billExpenses = new Bill();
        billExpenses.setUserId(processInstance.getExecutorId());
        billExpenses.setBillType(0);
        billExpenses.setTradeType(0);
        billExpenses.setBillNumber(productOrder.getRealAmount());
        billExpenses.setOrderNo(productOrder.getNo());
        billExpenses.setBillNo(billNo);
        bills.add(billIncome);
        bills.add(billExpenses);
        return bills;
    }
}
